#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：bignet_sumo_dev
@File ：main-dev-tongluoshan.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/3/24 9:40
'''

from __future__ import absolute_import
from __future__ import print_function

import asyncio
import json
import math
import time
from collections import defaultdict
from multiprocessing import Process, Queue

import numpy
import sumolib
import traci.constants as tc
from kafka import KafkaProducer, KafkaConsumer
from loguru import logger

from config.config_kakou import GLOBAL_POLYGON_CENTER, MAX_SPEED, KAFKA_HOST, \
    RADAR_DATA_LIST, USE_LIBSUMO, USE_GUI, KAFKA_E1_LIST, \
    ENTER_GANTRY_INFO, EXIT_RAMP_INFO
from utils import sumo_utils
from utils.plate_utils import is_vehicle_passing_checkpoint, extract_and_convert
from utils.pursuit_utils import calculate_catch_down_time, haversine, calculate_catch_up_time
from utils.tools import calculate_distance, get_timestamp_milliseconds, time_str
from websocket.websocketserver1 import WebSocketServer

if USE_LIBSUMO:
    import libsumo as traci
else:
    import traci

logger.add("/sumo/debug.log", format="{time} {level} {message}", level="DEBUG")


def senf_kafka_e1(q_s, p_id):
    kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                   value_serializer=lambda x: json.dumps(x).encode('utf-8'),
                                   compression_type='gzip')
    try:
        simu_json_list_before = None
        send_time = 0
        send_count = 0
        frame_num = 0
        # websocket信息
        ws_server = WebSocketServer()
        ws_server.start()
        loop = asyncio.new_event_loop()  # 创建新的事件循环
        asyncio.set_event_loop(loop)
        while True:
            t1 = time.time()
            frame_num += 1
            simu_json_list = None
            while not q_s.empty():
                simu_json_list = q_s.get()
                # print("发送端", len(simu_json_list))
            if simu_json_list:
                simu_json_list_before = simu_json_list
            if simu_json_list_before:
                # logger.info(f"感知数量为{len(one_list)} 仿真数量为{len(self.simu_json_list_before)}")
                # kafka_producer.send(KAFKA_E1_TOPIC_SIMU, value=simu_json_list_before,
                #                     key=str(int(time.time())))
                future = asyncio.run_coroutine_threadsafe(ws_server.broadcast(str(simu_json_list).replace("'", '"')),
                                                          ws_server.loop)
                future.result()  # 等待协程完成
            end_time = time.time()
            if end_time - send_time > 1:
                logger.info(f"进程{p_id} 发送帧率 {abs(send_count - frame_num)}  发送队列长度{q_s.qsize()}")
                send_time = end_time
                send_count = frame_num = 0
            if end_time - t1 < 0.1:
                time.sleep(0.1 - (end_time - t1))
    except Exception as e:
        logger.error(f"数据发送异常: {e}")


def read_trafficE1_channel(target_channel, q_r, kafka_queue):
    # 实时接收全域E1帧数据
    kafka_consumer_e1 = KafkaConsumer(
        target_channel,
        bootstrap_servers=KAFKA_HOST,
        auto_offset_reset='latest',
        enable_auto_commit=False
    )
    logger.info(f"全域E1 {target_channel} 监听已就绪")
    kafka_queue.put([target_channel, int(time.time())])
    while True:
        time.sleep(0.1)
        for message in kafka_consumer_e1:
            if message:
                try:
                    message_value = message.value.decode('utf-8')
                    data = json.loads(message_value)
                    q_r.put(data.copy())
                except Exception as e:
                    logger.error(f"数据接收异常：{e}")
                    continue


class SumoMain:
    def __init__(self, sumo_net_file, sumo_cfg_file, gui, process_id):
        """
        仿真初始化
        :param sumo_net_file:  仿真路网文件路径
        :param sumo_cfg_file:  仿真配置文件路径
        :param gui: 是否开启gui
        :param process_id: 进程id
        """
        self.sumo_net_file = sumo_net_file
        self.sumo_cfg_file = sumo_cfg_file
        self.gui = gui
        self.process_id = process_id
        # 仿真路网对象
        self.sumo_net_instance = None
        self.init_sumo()
        self.simu_json_list = []
        self.simu_json_list_before = []

        # 区间加减速车辆
        self.interval_tran_car = defaultdict(lambda: defaultdict(dict))
        self.sim_speed_cache_dict = dict()
        self.car_in_sn_dict = dict()
        self.global_vehicles = {}
        self.batch_num = 0
        self.global_time = time.time()
        self.global_num = 0
        self.statistic_info = {}
        # 进程控制
        self.son_process = {}

    def init_sumo(self):
        """
        初始化sumo  加载
        :return:
        """
        # 加载路网
        self.sumo_net_instance = sumolib.net.readNet(self.sumo_net_file, withInternal=True, withLatestPrograms=True)
        # 启动仿真
        sumo_utils.start_sumo(self.sumo_cfg_file, False, gui=self.gui)
        # 设置所有车道的最大通行速度
        for id in traci.lane.getIDList():
            traci.lane.setMaxSpeed(id, MAX_SPEED)
        # 新增区域中心矩形，用于订阅100KM范围内的所有仿真车辆信息，并配置监听信息内容
        traci.polygon.add('GLOBAL_POLYGON_CENTER', GLOBAL_POLYGON_CENTER, (1, 0, 1))
        traci.polygon.subscribeContext('GLOBAL_POLYGON_CENTER',
                                       tc.CMD_GET_VEHICLE_VARIABLE, 1000000,
                                       [tc.VAR_SPEED, tc.VAR_TYPE, tc.VAR_POSITION, tc.VAR_ANGLE, ])
        # 将雷达、卡口相机的经纬度和终点位置转为路网上所在的路段id 位置距离 车道号，起始点和终点所在的路段id添加路由
        for sn, sn_info in RADAR_DATA_LIST.items():
            # 转换经纬度到路网坐标
            center_x, center_y = traci.simulation.convertGeo(sn_info['lon'], sn_info['lat'], fromGeo=True)
            # 创建圆形区域，表示相机或雷达的位置，用于在使用gui可视化时，展示设备具体位置
            if sn_info['normal']:
                # 创建圆形区域
                self.create_circle_polygon(sn, center_x, center_y, radius=50)
            else:
                self.create_circle_polygon(sn, center_x, center_y, radius=50, color=(255, 0, 0, 255))

    def create_circle_polygon(self, vehID, center_x, center_y, radius, color=(0, 255, 0, 255), filled=True):
        """
        在指定位置创建一个近似圆形的多边形。

        :param vehID: 车辆 ID 或其他唯一标识符，用于命名多边形
        :param center_x: 圆心的 x 坐标
        :param center_y: 圆心的 y 坐标
        :param radius: 圆的半径
        :param color: 多边形的颜色 (R,G,B,A)，默认绿色不透明
        :param filled: 是否填充多边形，默认 True
        """
        # 生成圆周上的点
        num_points = 50  # 点的数量越多，圆越光滑
        points = []
        for i in range(num_points):
            angle = 2 * math.pi * i / num_points
            x = center_x + radius * math.cos(angle)
            y = center_y + radius * math.sin(angle)
            points.append((x, y))
        # 添加多边形
        polygon_id = f"circle_{vehID}"
        traci.polygon.add(polygon_id, shape=points, color=color, fill=filled, layer=0)

    def control_car(self):
        global_vehicles = self.global_vehicles.copy()
        car_in_sn_dict = self.car_in_sn_dict.copy()
        """车辆发车控制"""
        while not re_queue_data.empty():
            try:
                vehicle_info = re_queue_data.get()
                logger.info(f"仿真过程中接收到车辆信息：{vehicle_info}")
                if vehicle_info:
                    # 判断当前数据处于哪个门架
                    global_id = vehicle_info['global_id']
                    station = vehicle_info['station']
                    v_lon = vehicle_info['lon']
                    v_lat = vehicle_info['lat']
                    lane = vehicle_info['lane']
                    timestamp = get_timestamp_milliseconds(vehicle_info['timestamp'])
                    v_speed = vehicle_info['speed'] / 3.6
                    edge_id, pos, real_lane_index = traci.simulation.convertRoad(v_lon, v_lat, True)
                    # self.interval_real_car[station][global_id] = vehicle_info
                    # 如果station在入口门架中，延迟发车
                    if station in ENTER_GANTRY_INFO:
                        # 车辆进行发车
                        self.add_camera_car(vehicle_info, station)
                    else:
                        # 判断车辆是否已经在路网上
                        if global_id in global_vehicles.keys():
                            # 普通点位 只做加减速控制
                            veh_info = global_vehicles.get(global_id)
                            pos = veh_info[tc.VAR_POSITION]
                            ang = veh_info[tc.VAR_ANGLE]
                            sim_speed = veh_info[tc.VAR_SPEED]
                            lane_index = 0
                            car_in_sn_info = car_in_sn_dict.get(global_id)
                            sim_lon, sim_lat = traci.simulation.convertGeo(pos[0], pos[1], fromGeo=False)
                            # 判断该门架是否是出口匝道
                            if station in EXIT_RAMP_INFO:
                                arrive_edge_id = RADAR_DATA_LIST.get(station)['arrive_edge_id']
                                if not self.set_route_to_edge(global_id, arrive_edge_id):
                                    # x, y = traci.simulation.convertGeo(v_lon, v_lat, fromGeo=True)
                                    # edge_id, pos, lane_index = traci.simulation.convertRoad(v_lon, v_lat, True)
                                    # traci.vehicle.moveToXY(vehID=global_id, edgeID=edge_id, laneIndex=lane_index, x=x,
                                    #                        y=y, angle=-1.0, keepRoute=1)
                                    traci.vehicle.remove(global_id)
                                    self.add_camera_car(vehicle_info, station)
                            else:
                                # 判断车辆是否已经经过门架
                                pass_flag = is_vehicle_passing_checkpoint(sim_lon, sim_lat, ang, v_lon, v_lat)
                                # 如果经过，判断车辆此时与门架当前的位置差
                                if pass_flag:
                                    result = calculate_catch_down_time(sim_speed, v_speed, 10, 20,
                                                                       sim_lon, sim_lat, v_lon, v_lat)
                                    t_acc, t_const = result
                                    v_lon = vehicle_info['lon']
                                    v_lat = vehicle_info['lat']
                                    _distance = haversine(v_lon, v_lat, sim_lon, sim_lat)
                                    logger.info(
                                        f'{global_id} 车辆经过门架： {station}，{v_lon}, {v_lat}, {sim_lon},{sim_lat}'
                                        f' 距离门架距离为{_distance}米，当前速度为：{sim_speed},'
                                        f' 断面速度为  {v_speed} 做减速，减速时间为：{t_acc}  持续时间为：{t_const}')
                                    self.interval_tran_car[station][global_id][timestamp] = 1
                                    # 如果距离过小，只需要调整速度即可
                                    if _distance < 20:
                                        traci.vehicle.setSpeed(global_id, v_speed)
                                    # elif _distance > 1000:
                                    #     # 超过1km 直接跳变,断面发车
                                    #     traci.vehicle.remove(global_id)
                                    #     self.add_camera_car(vehicle_info, station)
                                    # else:
                                    if t_const:
                                        # 车辆进行减速
                                        traci.vehicle.slowDown(global_id, sim_speed * 0.8, t_acc)
                                        car_in_sn_info["match_sn"] = station
                                        self.car_in_sn_dict[global_id] = car_in_sn_info
                                        self.sim_speed_cache_dict[global_id] = {
                                            'sn': station,
                                            't_acc': t_acc * 1000,
                                            't_const': t_const * 1000,
                                            'time': int(time.time() * 1000),
                                            'i_speed': v_speed,
                                            'flag': 999
                                        }
                                    else:
                                        traci.vehicle.slowDown(global_id, sim_speed * 0.8, 5)
                                    self.control_lane_change(global_id, lane_index, real_lane_index)
                                else:
                                    result = calculate_catch_up_time(sim_speed, v_speed, 5, 20,
                                                                     sim_lon, sim_lat, v_lon, v_lat)
                                    t_acc, t_const = result
                                    _distance = haversine(v_lon, v_lat, sim_lon, sim_lat)
                                    logger.info(
                                        f'{global_id} 车辆未经过门架： {station}，{v_lon}, {v_lat}, {sim_lon},{sim_lat} '
                                        f'距离门架距离为{_distance}米，当前速度为：{sim_speed},'
                                        f' 断面速度为  {v_speed} 做加速，加速时间为：{t_acc}  持续时间为：{t_const}')
                                    if _distance < 20:
                                        traci.vehicle.setSpeed(global_id, v_speed)
                                        self.control_lane_change(global_id, lane_index, real_lane_index)
                                        self.interval_tran_car[station][global_id][timestamp] = 0
                                    elif _distance > 1000:
                                        # x, y = traci.simulation.convertGeo(v_lon, v_lat, fromGeo=True)
                                        # edge_id, pos, lane_index = traci.simulation.convertRoad(v_lon, v_lat, True)
                                        # traci.vehicle.moveToXY(vehID=global_id, edgeID=edge_id, laneIndex=lane_index, x=x,
                                        #                        y=y, angle=-1.0, keepRoute=1)
                                        traci.vehicle.remove(global_id)
                                        self.add_camera_car(vehicle_info, station)
                                    else:
                                        if t_const:
                                            # 说明能追上
                                            traci.vehicle.slowDown(global_id, sim_speed * 1.2, t_acc)
                                            car_in_sn_info = self.car_in_sn_dict.get(global_id)
                                            car_in_sn_info["match_sn"] = station
                                            self.car_in_sn_dict[global_id] = car_in_sn_info
                                            self.sim_speed_cache_dict[global_id] = {
                                                'sn': station,
                                                't_acc': t_acc * 1000,
                                                't_const': t_const * 1000,
                                                'time': int(time.time() * 1000),
                                                'i_speed': v_speed,
                                                'flag': 999
                                            }
                                            self.interval_tran_car[station][global_id][timestamp] = 2
                                        else:
                                            traci.vehicle.slowDown(global_id, sim_speed * 1.2, 5)
                                        self.control_lane_change(global_id, lane_index, real_lane_index)
            except Exception as e:
                logger.error(f'获取车辆信息失败：{e}')

    def control_info_car(self):
        """车辆发车控制"""
        # 速度控制
        all_vehicle_keys = self.global_vehicles.copy()
        sim_speed_cache_dict = self.sim_speed_cache_dict.copy()
        for global_car_id, sim_speed_cache in sim_speed_cache_dict.items():
            t_acc = sim_speed_cache['t_acc']
            t_const = sim_speed_cache['t_const']
            s_time = sim_speed_cache['time']
            i_speed = sim_speed_cache['i_speed']
            sn = sim_speed_cache['sn']
            n_time = time.time() * 1000
            if n_time - s_time - t_acc - t_const > 0:
                if global_car_id in all_vehicle_keys:
                    # 速度变化后，移除车辆
                    traci.vehicle.setSpeed(global_car_id, i_speed)
                    logger.info(f'经过门架：{sn}, 车辆{global_car_id} 速度变化结束，恢复初始速度{i_speed}')
                    self.sim_speed_cache_dict.pop(global_car_id)

    def get_statisticInfo(self):

        global_vehicles = self.global_vehicles.copy()
        statisticInfo = {}
        statisticInfo['totalNum'] = len(global_vehicles)
        type_count_dict = {}
        speed_list = []
        for vehicle_id, veh_info in dict(global_vehicles).items():
            speed = veh_info[tc.VAR_SPEED]
            speed_list.append(speed)
            try:
                etc_info = eval(traci.vehicle.getParameter(vehicle_id, "etc_info"))
            except Exception as e:
                continue
            carType = etc_info['carType']
            if carType not in type_count_dict.keys():
                type_count_dict[carType] = 1
            else:
                type_count_dict[carType] = type_count_dict[carType] + 1
        meanSpeed = numpy.mean(speed_list) * 3.6 if len(speed_list) > 0 else 0
        totalNumTypeInfo = []
        for t, c in type_count_dict.items():
            totalNumTypeInfo.append({'carType': t, 'carNum': c})
        statisticInfo['meanSpeed'] = round(meanSpeed, 1)
        statisticInfo['totalNumTypeInfo'] = totalNumTypeInfo
        self.statistic_info = statisticInfo

    def set_route_to_edge(self, vehicle_id, target_edge):
        """
        直接将车辆路由设置为到达目标边的最短路径
        :param vehicle_id: 车辆ID
        :param target_edge: 目标边ID
        :return: 是否设置成功
        """
        if vehicle_id not in traci.vehicle.getIDList():
            print(f"车辆 {vehicle_id} 不存在")
            return False
        current_edge = traci.vehicle.getRoadID(vehicle_id)
        try:
            # 直接获取到目标边的路径
            route = traci.simulation.findRoute(current_edge, target_edge)
            # 如果路径存在且有效
            if route and route.edges:
                traci.vehicle.setRoute(vehicle_id, route.edges)
                print(f"车辆 {vehicle_id} 路由已更新: {route.edges}")
                return True
            else:
                print(f"无法找到从 {current_edge} 到 {target_edge} 的路径")
                return False
        except traci.TraCIException as e:
            print(f"路由设置错误: {e}")
            return False

    def control_lane_change(self, global_id, now_lane, targe_lane):
        if now_lane != targe_lane:
            traci.vehicle.changeLaneRelative(global_id, targe_lane, 3)

    def control_delete_car(self):
        """删车控制"""
        # 如果连续3个没有匹配上 则删除车辆
        car_in_sn_dict = self.car_in_sn_dict.copy()
        for car_id, car_info in list(car_in_sn_dict.items()):
            match_sn = car_info['match_sn']
            now_sn = car_info['now_sn']
            # print(car_in_sn_dict)
            for lidar__path in self.lidar__paths:
                if match_sn in lidar__path:
                    match_sn_index = lidar__path.index(match_sn)
                    now_sn_index = lidar__path.index(now_sn)
                    # print(via_sn_list)
                    if abs(now_sn_index - match_sn_index) > 2:
                        logger.info(f'车辆{car_id} 删除，未经过门架：{match_sn}，已经过门架：{now_sn}')
                        traci.vehicle.remove(car_id)
                        self.car_in_sn_dict.pop(car_id)

    def add_camera_car(self, vehicle_info, sn):
        plate_global_id = vehicle_info['global_id']

        set_speed = vehicle_info['speed'] / 3.6 if 30 <= vehicle_info['speed'] <= 140 else 80 / 3.6
        edge_id, pos, lane_index = traci.simulation.convertRoad(vehicle_info['lon'],
                                                                vehicle_info['lat'],
                                                                True)
        lane_index = vehicle_info['lane']
        route_ids = traci.route.getIDList()
        if edge_id not in route_ids:
            arrive_edge_id = RADAR_DATA_LIST.get(sn)["arrive_edge_id"]
            route = traci.simulation.findRoute(edge_id, arrive_edge_id)
            # 如果路径存在且有效
            if route and route.edges:
                traci.route.add(edge_id, [str(edge_id), arrive_edge_id])
            else:
                logger.info(f"发车失败，无法找到从 {edge_id} 到 {arrive_edge_id} 的路径")
        try:
            traci.vehicle.add(
                vehID=str(plate_global_id),
                routeID=edge_id,
                departPos=pos,
                typeID=1, depart='now',
                departSpeed=str(set_speed),
                departLane=lane_index)
        except Exception as e:
            logger.info(f"加车错误{e}  {lane_index}")
            if 'Invalid departlane' in str(e):
                traci.vehicle.add(
                    vehID=str(plate_global_id),
                    routeID=edge_id,
                    departPos=pos,
                    typeID=1, depart='now',
                    departSpeed=str(set_speed),
                    departLane=lane_index)
        traci.vehicle.setSpeed(str(plate_global_id), set_speed)
        lon = RADAR_DATA_LIST[sn]['lon']
        lat = RADAR_DATA_LIST[sn]['lat']
        v_lon = vehicle_info['lon']
        v_lat = vehicle_info['lat']
        dis = calculate_distance(lon, lat, v_lon, v_lat)
        print(str(vehicle_info['picLicense']), edge_id, pos, set_speed, "发车距离门架：", dis, "米")
        traci.vehicle.setParameter(str(plate_global_id), "etc_info",
                                   str(vehicle_info))
        # 计算发车位置距离门架多远
        self.car_in_sn_dict[plate_global_id] = {
            "start_sn": sn,
            "now_sn": sn,
            "match_sn": sn,
            "timestamp": int(time.time())
        }

    def interval_speed_control(self):
        pass

    def car_info_handle(self, global_vehicles, queue_data):
        try:
            simu_json_list = []
            objs = []
            sim_speed_cache_dict = self.sim_speed_cache_dict.copy()
            for vehicle_id, veh_info in dict(global_vehicles).items():
                pos = veh_info[tc.VAR_POSITION]
                longitude, latitude = traci.simulation.convertGeo(pos[0], pos[1], fromGeo=False)
                course_angle = veh_info[tc.VAR_ANGLE]
                speed = veh_info[tc.VAR_SPEED]
                # 判断当前行为
                flag = 1
                if vehicle_id in sim_speed_cache_dict.keys():
                    sim_speed_info = sim_speed_cache_dict.get(vehicle_id)
                    flag = sim_speed_info['flag']
                etc_info = eval(traci.vehicle.getParameter(vehicle_id, "etc_info"))
                r_id = extract_and_convert(vehicle_id)
                # decoded = base64.b64decode(vehicle_id).decode("utf-8")
                # 出京 org code 定义为 CJ-10-SUM
                # 进京 org code 定义为 JJ-10-SUM
                temp = {"orgCode": "simu", "id": r_id, "carType": etc_info['carType'], "carColor": 0,
                        "picLicense": etc_info['picLicense'], "lon": longitude, "lat": latitude,
                        "altitude": 0, "speed": speed * 3.6, "angle": round(course_angle, 1),
                        "station": etc_info['station'], "lane": etc_info['lane'],
                        "enGap": 0, "stakeNum": "0", "speedAvg": 0, "passTime": 0,
                        "firstReceiveTime": time_str(int(time.time() * 1000)), "disBefore": 0,
                        "licenseColor": 9}
                obj = {'id': vehicle_id, 'type': etc_info['carType'], 'source': etc_info['station'],
                       'laneNum': etc_info['lane'], 'longitude': longitude, 'latitude': latitude,
                       'speed': speed * 3.6, 'courseAngle': course_angle,
                       'plate': etc_info['picLicense'], 'flag': flag, 'orgCode': 'JJ-10-01'}
                simu_json_list.append(temp)
                objs.append(obj)
            if self.batch_num % 10 == 0:
                self.get_statisticInfo()
            # 出京 org code 定义为 CJ-10-SUM
            # 进京 org code 定义为 JJ-10-SUM
            ws_data = {"frameNum": self.batch_num, "globalTime": time_str(int(time.time() * 1000)), "obj": objs}
            frame_info = {"orgCode": "simu", "frameNum": self.batch_num,
                          "globalTime": time_str(int(time.time() * 1000)),
                          "targetList": simu_json_list, "statisticInfo": self.statistic_info, }
            data_json = json.dumps(frame_info)
            queue_data.put(data_json)
        except Exception as e:
            print(e)

    def history_data_clean(self):
        try:
            if self.batch_num % 10:
                car_in_sn_dict = self.car_in_sn_dict.copy()
                interval_tran_car = self.interval_tran_car.copy()
                sim_speed_cache_dict = self.sim_speed_cache_dict.copy()
                for global_id in car_in_sn_dict.keys():
                    car_in_sn_info = car_in_sn_dict.get(global_id)
                    timestamp = car_in_sn_info['timestamp']
                    if int(time.time()) - timestamp > 60 * 60:
                        del self.car_in_sn_dict[global_id]
                for global_id in sim_speed_cache_dict.keys():
                    sim_speed_cache_info = sim_speed_cache_dict.get(global_id)
                    start_time = int(sim_speed_cache_info['time'])
                    if int(time.time() * 1000) - start_time > 60 * 60 * 1000:
                        del self.sim_speed_cache_dict[global_id]
                    # 遍历站点层级
                for station, global_ids in list(interval_tran_car.items()):
                    # 遍历全局ID层级
                    for global_id, timestamps in list(global_ids.items()):
                        # 遍历时间戳层级，删除过期数据
                        expired_timestamps = [ts for ts in timestamps if
                                              int(time.time() * 1000) - int(ts) > 60 * 60 * 1000]
                        if len(expired_timestamps) > 0:
                            self.interval_tran_car.pop(global_id)
        except Exception as e:
            logger.error(f"数据清理异常{e}")

    def main_loop(self, stop_data, time_data):
        global re_queue_data, kafka_queue

        kafka_queue = Queue(maxsize=2000)
        # 在子进程中初始化多进程队列和资源
        queue_data = Queue(maxsize=2000)  # 多进程队列在此处创建
        # Process(target=send_kafka_1, args=(queue_data, self.process_id,)).start()
        # 在子进程中初始化多进程队列和资源
        re_queue_data = Queue(maxsize=2000)  # 多进程队列在此处创建
        kafka_time_dict = {}

        kafka_time_dict[KAFKA_E1_LIST] = int(time.time())
        son_process = Process(target=read_trafficE1_channel,
                              args=(KAFKA_E1_LIST, re_queue_data, kafka_queue), daemon=True)
        son_process.start()
        self.son_process[KAFKA_E1_LIST] = son_process
        son_process = Process(target=senf_kafka_e1,
                              args=(queue_data, self.process_id), daemon=True)
        son_process.start()
        self.son_process['send'] = son_process
        low_frame_num = 0
        batch_num = 0
        all_set = set()

        while True:
            # try:
            start_time = time.time()
            start_time_m = int(time.time() * 1000)

            # ================================收到门架上报数据，进行车辆处理==========================================
            self.batch_num = batch_num
            global_vehicles = traci.polygon.getContextSubscriptionResults('GLOBAL_POLYGON_CENTER')
            self.global_vehicles = global_vehicles
            # 1小时前数据清理
            self.history_data_clean()
            # 发车车辆控制
            self.control_car()
            # 结束速度控制
            self.control_info_car()
            # self.control_delete_car()
            # 区间速度控制
            # self.interval_speed_control()
            # 发车处理
            self.car_info_handle(global_vehicles, queue_data)

            batch_num += 1
            traci.simulationStep()
            end_time = time.time()
            all_set.update(set(global_vehicles.keys()))
            # logger.info(f"仿真数量总数{len(all_set)}")
            if end_time - self.global_time > 1:
                while kafka_queue.qsize() > 0:
                    kafka_info = kafka_queue.get()
                    kafka_topic = kafka_info[0]
                    ts = kafka_info[1]
                    kafka_time_dict[kafka_topic] = ts

                if abs(kafka_time_dict[KAFKA_E1_LIST] - end_time) > 60 * 5:
                    if KAFKA_E1_LIST in self.son_process.keys():
                        logger.info(f"topic为{KAFKA_E1_LIST} 的接收进程,超过五分钟未收到数据，重启服务")
                        s_p = self.son_process[KAFKA_E1_LIST]
                        s_p.terminate()
                        s_p.join()
                        son_process = Process(target=read_trafficE1_channel,
                                              args=(KAFKA_E1_LIST, re_queue_data, kafka_queue), daemon=True)
                        son_process.start()
                        self.son_process[KAFKA_E1_LIST] = son_process

                time_data.put(int(time.time()))
                simu_frame = abs(self.global_num - batch_num)
                logger.info(f"进程{self.process_id} 仿真帧率 {simu_frame} 仿真数量{len(global_vehicles)}")
                if simu_frame < 7:
                    low_frame_num += 1
                if low_frame_num > 10 or (self.global_num > 8000 and len(global_vehicles) == 0):
                    logger.info(f"关闭子进程")
                    for s_name, s_p in self.son_process.items():
                        s_p.terminate()
                        s_p.join()
                    stop_data.put(1)
                self.global_time = end_time
                self.global_num = batch_num
            if start_time and (end_time - start_time) < 0.1:
                # print("耗时", (0.1 - (end_time - start_time)))
                time.sleep(0.1 - (end_time - start_time))
            else:
                if start_time:
                    # logger.info(f"单帧耗时大于0.1s {end_time - start_time}")
                    pass
            # except Exception as e:
            #     logger.error(f"主循环异常{str(e)}")
            #     continue


def run_sumo_simulation(sumo_net_file, sumo_cfg_file, gui, process_id, stop_data, time_data):
    sumo_main = SumoMain(sumo_net_file, sumo_cfg_file, gui, process_id)
    sumo_main.main_loop(stop_data, time_data)


if __name__ == '__main__':
    # # 路网文件
    # sumo_net_file = "/sumo/net/jingtai.net.xml0611"
    # # 配置文件
    # sumo_cfg_file = "/sumo/net/jingtai.sumocfg"
    # 路网文件
    sumo_net_file = "./net/jingtai.net.xml"
    # 配置文件
    sumo_cfg_file = "./net/jingtai.sumocfg"
    # 是否启用gui可视化界面，调试阶段启用，线上部署关闭减少资源占用
    stop_data = Queue(maxsize=2000)  # 创建队列，仿真循环里面判断，如果仿真帧率小于7或者仿真路网上仿真车流数量为零，向队列中添加数据
    time_data = Queue(maxsize=2000)  # 创建队列，仿真主循环里一秒向里面添加一次系统时间戳
    gui = USE_GUI
    # tess_process = Process(target=run_sumo_simulation,
    #                        args=(sumo_net_file, sumo_cfg_file, gui, 1, stop_data, time_data))
    # tess_process.start()
    # time_simu = int(time.time())
    for i in range(1):
        logger.info("开启线程")
        tess_process = Process(target=run_sumo_simulation,
                               args=(sumo_net_file, sumo_cfg_file, gui, i, stop_data, time_data))
        tess_process.start()
        time_simu = int(time.time())
        while True:
            stop_flag = False
            # 如果仿真的最新时间与当前系统的时间的时间差大于一分钟，认为仿真异常关闭，需要重启服务
            while time_data.qsize() > 0:
                time_simu = time_data.get()
            if time_simu:
                # logger.info(f"时间差{abs(time_simu - int(time.time()))}")
                if abs(time_simu - int(time.time())) > 60:
                    stop_flag = True
                    time_simu = int(time.time())
            # 如果仿真帧率小于7或者仿真路网上仿真车流数量为零，认为仿真异常，需要重启服务
            while stop_data.qsize() > 0:
                _ = stop_data.get()
                stop_flag = True
            if stop_flag:
                logger.info("关闭开启线程")
                tess_process.terminate()
                tess_process.join()
                logger.info("重新开启线程")
                tess_process = Process(target=run_sumo_simulation,
                                       args=(sumo_net_file, sumo_cfg_file, gui, i, stop_data, time_data))
                tess_process.start()
            time.sleep(1)
